package com.pachiraframework.springcloud.spark.service;

import java.io.Serializable;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.util.LongAccumulator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import scala.Tuple2;

@Component
public class WordCountService implements Serializable {
	private static final Pattern SPACE = Pattern.compile(" ");

	 @Autowired
	 private JavaSparkContext sc;

	 //总是报Serializable异常，如果不用配置JavaSparkContext的就没事
	public Map<String, Integer> run() {
		
		
		
//		SparkConf conf = new SparkConf().setAppName("Accumulator").setMaster(
//				"local");
//		JavaSparkContext sc = new JavaSparkContext(conf);

		// 创建Accumulator变量
		// 需要调用SparkContext的accumulator()方法
//		final Accumulator<Integer> sum = sc.accumulator(0);

//		final LongAccumulator sum = sc.sc().longAccumulator("jishu");
//		
//		List<Integer> numberList = Arrays.asList(new Integer(1), new Integer(2), new Integer(3), new Integer(4), new Integer(5));
//		JavaRDD<Integer> numbers = sc.parallelize(numberList);
//
//		numbers.foreach(new VoidFunction<Integer>() {
//
//			private static final long serialVersionUID = 1L;
//
//			public void call(Integer t) throws Exception {
//				// 然后在函数内部，就可以对Accumulator变量，调用add()方法，累加值
//				sum.add(t);
//			}
//
//		});
//
//		// 在driver程序中，可以调用Accumulator的value()方法，获取其值
//		System.out.println(sum.value());
//
//		sc.close();

		return null;

	}
}
